package org.zyj.io;

import lombok.SneakyThrows;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Echo Server 4
 * 第四代: 使用Selector的NIO
 * 1个boss线程专门处理accept事件
 * 4个worker线程专门处理读写事件
 * ---
 * 优点:
 * 1. 1个线程专门处理ACCEPT事件, 对新连接的响应非常快
 * 2. 专门有4个线程(可以改), 来处理读写事件, 效率非常的高
 * 3. 思想上, 这种设计是能够最大的压榨机器性能的
 * ---
 * 缺点:
 * 1. 需要自己直接使用NIO的接口, 那么意味着可能会遇到各种坑
 * 2. 已经有Netty这种成熟框架包装了NIO的接口, 没有必要自己再写一套基于NIO的框架(但是方便帮助自己理解netty的思想)
 */
public class SelectorNIO_2 {


    public static void main(String[] args) throws IOException, InterruptedException {
        //new的时候,就创建了boss线程和worker线程,只是没有start
        SocketMultiplexingThreads socketMultiplexingThreads = new SocketMultiplexingThreads(9090, 4);
        //start boss线程和worker线程, 此时可以接收请求了
        socketMultiplexingThreads.start();
    }

    //包装了boss线程和worker线程的创建初始化过程
    static class SocketMultiplexingThreads {

        //服务
        private ServerSocketChannel server;

        //boss线程
        private BossThread bossThread;

        //worker线程
        private List<WorkerThread> workerThreads;

        //服务绑定的端口号
        private int port;

        //构造方法, 同时初始化server, bossThread, workerThread
        public SocketMultiplexingThreads(int port, int workerNum) throws IOException {
            this.port = port;
            server = ServerSocketChannel.open();
            server.configureBlocking(false);//非阻塞
            server.bind(new InetSocketAddress(this.port));
            this.bossThread = new BossThread(server, this);
            //创建boss线程
            workerThreads = new ArrayList<>(workerNum);
            for (int i = 0; i < workerNum; i++) {
                //创建worker线程
                workerThreads.add(new WorkerThread(server));
            }
        }

        //粗糙的启动方法
        public void start() throws InterruptedException {
            bossThread.start();
            for (WorkerThread workerThread : workerThreads) {
                workerThread.start();
            }
            Thread.sleep(1000 * 60 * 60);
        }

        //随机获取一个worker线程
        WorkerThread randomGetWorkerThread() {
            Random random = new Random();
            int index = random.nextInt(workerThreads.size());
            return workerThreads.get(index);
        }
    }

    //boss线程只处理accept事件
    static class BossThread extends Thread {

        private Selector bossSelector;
        private SocketMultiplexingThreads socketMultiplexingThreads;

        public BossThread(ServerSocketChannel server, SocketMultiplexingThreads socketMultiplexingThreads) throws IOException {
            this.socketMultiplexingThreads = socketMultiplexingThreads;
            //创建Boss多路复用器
            this.bossSelector = Selector.open();
            //向主线程注册ACCEPT事件
            server.register(bossSelector, SelectionKey.OP_ACCEPT);//向多路复用器 注册server, 并且设置监听ACCEPT事件
            System.out.println("Boss 线程创建完毕");
        }

        @lombok.SneakyThrows
        @Override
        public void run() {
            //boss线程不停的轮询, 等待新的连接, 专门处理accept事件
            while (true) {
                int readyNum = bossSelector.select(100);
                if (readyNum > 0) {
                    Set<SelectionKey> selectionKeys = bossSelector.selectedKeys();//获取有事件的key, 这个实际上拿到的是多路复用器中的一个成员变量, 搞完只要要把key remove出去, 非常麻烦
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        System.out.println("boss 线程处理事件");
                        SelectionKey key = iterator.next();
                        iterator.remove();//这里需要remove掉, 如果不remove掉,下次循环还会遍历到这个key, 设计非常傻逼
                        //Boss线程只处理Accept事件
                        if (key.isAcceptable()) {
                            acceptHandler(key);//建立连接操作
                        }
                    }
                }
            }
        }

        //处理accept事件,将读事件绑定到worker线程上去
        private void acceptHandler(SelectionKey key) throws IOException {
            ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
            //1. 创建连接
            SocketChannel client = ssc.accept();//建立连接
            client.configureBlocking(false);
            //2. 分配buffer给这个连接, 其实这里如果把分配内存放到worker线程中去做, boss能处理更多的连接, 这里就不演示了, 不然代码就不清晰了
            ByteBuffer buffer = ByteBuffer.allocateDirect(8192);
            //3. 获取到worker线程的多路复用器
            WorkerThread workerThread = socketMultiplexingThreads.randomGetWorkerThread();//这里用的是随机获取一个, 算法可以自己实现
            Selector workerSelector = workerThread.getWorkerSelector();
            //4. 向多路复用器注册监听 该连接的读取事件
            SelectionKey selectionKey = client.register(workerSelector, SelectionKey.OP_READ, buffer);//向多路复用器绑定,读取事件
            //这个selectionKey和到时候发生读取事件读取的key是一个对象, 这里暂时没用它
            System.out.println("向worker线程注册读取事件: " + client.getRemoteAddress());
        }
    }

    //worker线程只处理读写事件
    static class WorkerThread extends Thread {

        private Selector workerSelector;
        private static AtomicInteger accumulator = new AtomicInteger(0);
        private int id = accumulator.getAndIncrement();//作为线程的标识

        public WorkerThread(ServerSocketChannel server) throws IOException {
            //创建Worker多路复用器
            this.workerSelector = Selector.open();
            System.out.println(this + " worker 线程创建完毕");
        }

        public Selector getWorkerSelector() {
            return this.workerSelector;
        }

        @SneakyThrows
        @Override
        public void run() {
            //worker线程不停的轮询, 等待新的连接, 专门处理读事件
            while (true) {
                int readyNum = workerSelector.select(100);
                if (readyNum > 0) {
                    Set<SelectionKey> selectionKeys = workerSelector.selectedKeys();//获取有事件的key, 这个实际上拿到的是多路复用器中的一个成员变量, 搞完只要要把key remove出去, 非常麻烦
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        System.out.println(this + " worker 线程处理事件");
                        SelectionKey key = iterator.next();
                        iterator.remove();//这里需要remove掉, 如果不remove掉,下次循环还会遍历到这个key, 设计非常傻逼
                        //Boss线程只处理Accept事件
                        if (key.isReadable()) {
                            readHandler(key);//建立连接操作
                        }
                    }
                }
            }
        }

        //worker线程处理读写事件
        private void readHandler(SelectionKey key) throws IOException {
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer buffer = (ByteBuffer) key.attachment();//这个就是在建立连接的时候,当做attachment存进去的buffer
            buffer.clear();
            int read;
            while (true) {
                read = client.read(buffer);
                if (read > 0) {
                    buffer.flip();
                    while (buffer.hasRemaining()) {
                        //这里是把数据写回客户端
                        //这里可以写其他的自己的业务逻辑
                        client.write(buffer);
                    }
                    buffer.clear();
                } else if (read == 0) {
                    break;
                } else { //-1 可能是客户端关闭了, close_wait
                    client.close();
                    break;
                }
            }
        }

        @Override
        public String toString() {
            return "worker线程[" + id + "]";
        }
    }


}
